package com.lds.memory.kafka.broker;

import com.lds.memory.kafka.enums.ErrorEnum;
import com.lds.memory.kafka.topic.Topic;
import com.lds.memory.kafka.topic.TopicConfig;
import com.lds.memory.kafka.topic.TopicGenerator;
import com.lds.memory.kafka.util.AssertUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * topic管理器
 * 创建、暂停、取消暂停、关闭（启动由topic自行管理）
 */
public class Broker {
    private final String name;
    private final Map<String, Topic> topicMap = new ConcurrentHashMap<>();

    public Broker() {
        this(null);
    }

    public Broker(String name) {
        this.name = StringUtils.isEmpty(name) ? "DEFAULT" : name;
    }

    public Topic createTopic(TopicConfig topicConfig) {
        Topic topic = TopicGenerator.generate(topicConfig);
        register(topic);
        return topic;
    }

    private void register(Topic topic) {
        AssertUtils.assertTrue(!topicMap.containsKey(topic.getName()), ErrorEnum.REPEAT_REGISTER_TOPIC);
        topicMap.put(topic.getName(), topic);
    }

    public void pause(String topicName) {
        Topic topic = topicMap.get(topicName);
        if (topic == null) return;

        topic.pause();
    }

    public void cancelPause(String topicName) {
        Topic topic = topicMap.get(topicName);
        if (topic == null) return;

        topic.cancelPause();
    }

    public void remove(String topicName) {
        Topic topic = topicMap.get(topicName);
        if (topic == null) return;

        topic.shutdown();

        topicMap.remove(topicName);
    }

}
